package data.platform.merge.internal.command;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import data.platform.common.service.command.MergeMessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.time.Duration;

@Service
@Slf4j
@RequiredArgsConstructor
public class MqMergeMessageServiceImpl implements MergeMessageService {

    @Value("${metric.merge.rabbitmq.exchange}")
    private String exchange;

    @Value("${metric.merge.rabbitmq.routingKey}")
    private String routingKey;

    @Value("${metric.merge.rabbitmq.delay}")
    private int delay;

    final Connection connection;

    Channel channel;

    @PostConstruct
    public void init() throws Exception {
        channel = connection.createChannel();
        channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
    }

    @Override
    public void sendMergeMessage(String message) {
        Mono.delay(Duration.ofMillis(delay * 1000))
                .subscribe(i -> {
                    try {
                        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                        builder = builder.contentType("text/plain")
                                .contentEncoding(StandardCharsets.UTF_8.name());
                        channel.basicPublish(exchange, routingKey, builder.build(), message.getBytes(StandardCharsets.UTF_8));
                        log.info("send merge message success");
                    } catch (Exception ex) {
                        log.error("", ex);
                    }
                });
    }
}
